package com.dx.flink.sink;

import com.alibaba.fastjson.JSONObject;
import com.dx.flink.dto.OAUser;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * 输出到im
 *
 * @author xufeng
 * @since 07.04.2023
 */
@Slf4j
@Data
@Component
public class ImSink extends RichSinkFunction<String> {

    Connection connection = null;
    PreparedStatement insertSmt = null;
    PreparedStatement deleteSmt = null;
    PreparedStatement updateSmt = null;


    @Override
    public void open(Configuration parameters) throws Exception {
        String url = "jdbc:mysql://192.168.2.253:3306/im?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false";
        connection = DriverManager.getConnection(url, "root", "123456");
        insertSmt = connection.prepareStatement("REPLACE into `user`(id,name) values (?,?)");
        deleteSmt = connection.prepareStatement("delete from `user` where id = ?");
        updateSmt = connection.prepareStatement("update  `user` set name = ? where id = ?");

    }

    @Override
    public void invoke(String value, Context context) throws Exception {

        //log.info("原始cdc数据 is {}",value);
        JSONObject jsonObject = JSONObject.parseObject(value);
        switch (jsonObject.get("type").toString()) {
            case "insert":
                OAUser afterData = JSONObject.parseObject(JSONObject.toJSONString(jsonObject.get("afterData")), OAUser.class);
                //直接执行更新语句
                insertSmt.setLong(1, afterData.getId());
                insertSmt.setString(2, afterData.getName());
                insertSmt.execute();
                log.info("执行新增语句成功 id is {}", afterData.getId());
                break;
            case "update":
                OAUser updateData = JSONObject.parseObject(JSONObject.toJSONString(jsonObject.get("afterData")), OAUser.class);
                //删除
                updateSmt.setString(1, updateData.getName());
                updateSmt.setLong(2, updateData.getId());
                updateSmt.execute();
                log.info("执行修改语句成功 id is {}", updateData.getId());
                break;
            case "delete":
                OAUser beforeData = JSONObject.parseObject(JSONObject.toJSONString(jsonObject.get("beforeData")), OAUser.class);
                //删除
                deleteSmt.setLong(1, beforeData.getId());
                deleteSmt.execute();
                log.info("执行删除语句成功 id is {}", beforeData.getId());
                break;
            default:
                break;
        }

    }

    @Override
    public void close() throws Exception {
        connection.close();
        insertSmt.close();
        updateSmt.close();
        insertSmt.close();
    }


}
